#include <iostream>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <atomic>

// 线程安全的队列
template<typename T>
class ThreadSafeQueue {
private:
    std::queue<T> queue_;
    mutable std::mutex mutex_;
    std::condition_variable cond_;
    std::atomic<bool> stop_flag_{false};

public:
    // 将数据推入队列
    void push(const T &value) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(value);
        cond_.notify_one(); // 通知一个等待的线程
    }

    // 尝试从队列弹出数据
    bool try_pop(T &value) {
        std::lock_guard<std::mutex> lock(mutex_);
        if (queue_.empty() || stop_flag_) {
            return false;
        }
        value = queue_.front();
        queue_.pop();
        return true;
    }

    // 等待并弹出数据
    bool wait_and_pop(T &value) {
        std::unique_lock<std::mutex> lock(mutex_);
        cond_.wait(lock, [this] { return !queue_.empty() || stop_flag_; });

        if (stop_flag_) {
            return false;
        }

        value = queue_.front();
        queue_.pop();
        return true;
    }

    // 停止队列操作
    void stop() {
        stop_flag_ = true;
        cond_.notify_all(); // 通知所有等待的线程
    }

    // 检查队列是否为空
    bool empty() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return queue_.empty();
    }
};


template<typename T>
class Worker {
private:
    ThreadSafeQueue<T> &queue_;
    std::function<void(const T &)> processor_;
    std::thread worker_thread_;
    std::atomic<bool> is_running_{false};

public:
    Worker(ThreadSafeQueue<T> &queue, std::function<void(const T &)> processor)
        : queue_(queue), processor_(std::move(processor)) {
    }

    ~Worker() {
        stop();
    }

    void start() {
        if (is_running_) return;

        is_running_ = true;
        worker_thread_ = std::thread([this]() {
            while (is_running_) {
                T value;
                if (queue_.wait_and_pop(value)) {
                    try {
                        processor_(value);
                    } catch (const std::exception &e) {
                        std::cerr << "Processing error: " << e.what() << '\n';
                    }
                } else {
                    // Queue was stopped
                    break;
                }
            }
        });
    }

    void stop() {
        if (!is_running_) return;

        is_running_ = false;
        queue_.stop();

        if (worker_thread_.joinable()) {
            //等待线程工作结束，queue.stop调用会限制线程执行最后一个任务后退出循环
            worker_thread_.join();
        }
    }
};


int thread_main() {
    ThreadSafeQueue<int> queue;

    // 定义处理函数
    auto processor = [](const int &value) {
        std::cout << "Processing value: " << value << '\n';
    };
    // 创建worker
    Worker<int> worker(queue, processor);
    worker.start();
    // 生产数据
    for (int i = 0; i < 10; ++i) {
        queue.push(i);
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    // 等待队列处理完成
    while (!queue.empty()) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    // 停止worker
    worker.stop();
    return 0;
}
